package com.zhanghe.study.study_scala.akka.spark.worker

import java.util.concurrent.TimeUnit

import akka.actor.{Actor, ActorSelection, ActorSystem, Props, Scheduler}
import com.typesafe.config.ConfigFactory
import com.zhanghe.study.study_scala.akka.spark.common.{HeartBeat, RegisterWorkerInfo, RegisteredWorkerInfo, SendHeartBeat}

import scala.concurrent.duration
import scala.concurrent.duration.FiniteDuration


/**
 * @author zh
 * @date 2021/5/10 16:38
 */
object SparkWorker {

  def main(args: Array[String]): Unit = {
    val (workerHost,workerPort,masterHost,masterPort) = ("127.0.0.1",10001,"127.0.0.1",10005)


    // 创建ActorSystem  remote配置
    val config = ConfigFactory.parseString(
      s"""
         |akka.actor.provider="akka.remote.RemoteActorRefProvider"
         |akka.remote.artery.canonical.hostname=$workerHost
         |akka.remote.artery.canonical.port=$workerPort
         |akka.actor.allow-java-serialization=on
            """.stripMargin)
    val sparkWorkerSystem = ActorSystem("SparkWorker", config)
    // 创建sparkWorker
    val sparkWorkerRef = sparkWorkerSystem.actorOf(Props(new SparkWorker(masterHost,masterPort)),"sparkWorker")

    // worker开始
    sparkWorkerRef.!("start")
  }

}

class SparkWorker(host:String,port:Int) extends Actor{

  // master的代理
  var masterProxy : ActorSelection = _

  // 生成worker的id
  val id = java.util.UUID.randomUUID().toString


  // 开始之前执行
  override def preStart(): Unit = {
    masterProxy = context.actorSelection(s"akka://SparkMaster@$host:$port/user/SparkMaster")
  }

  override def receive: Receive = {
    case "start" =>
      {
        println(s"worker--$id 开始工作了")
        masterProxy ! RegisterWorkerInfo(id,16,16*1024)
      }
    case RegisteredWorkerInfo =>{
      println("注册成功")
      // 注册成功之后，启动定时器，每隔一段时间发送SendHeartBeat
      import context.dispatcher
      context.system.scheduler.scheduleAtFixedRate(FiniteDuration(0,TimeUnit.MILLISECONDS),FiniteDuration(3000,TimeUnit.MILLISECONDS),self,SendHeartBeat)
    }
    case SendHeartBeat => {
      println(s"worker => $id 心跳发送")
      masterProxy ! HeartBeat(id)
    }

  }
}
